package com.lagou.scala.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

import scala.io.BufferedSource

object KafkaProducerForDruid {
  def main(args: Array[String]): Unit = {
    val brokers = "node51:9092,node52:9092,node53:9092"
    val topic = "lagoudruid2"
    val prop = new Properties()
    prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,brokers)
    prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,classOf[StringSerializer])
    prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,classOf[StringSerializer])
    val producer = new KafkaProducer[String,String](prop)

    val source: BufferedSource = scala.io.Source.fromFile("data/lagou_orders.json")
    val iter: Iterator[String] = source.getLines()
    iter.foreach{line =>
      val msg = new ProducerRecord[String, String](topic, line)
      producer.send(msg)
      Thread.sleep(10)
    }
    producer.close()
    source.close()
  }
}
